%%%-------------------------------------------------------------------
%%% @author wukai
%%% @copyright (C) 2018, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 13. 一月 2018 18:44
%%%-------------------------------------------------------------------
-module(himq_session).
-author("wukai").
-behaviour(gen_server2).
-include("himq.hrl").
-include_lib("stdlib/include/qlc.hrl").
-define(ONLINE,1).
-define(OFFLINE,0).
-export([session_list/2,push/1]).
-export([del/2, add/9]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
-record(state, {}).

init([]) ->
  ensure_table(),
  register(himq_session, self()),
  {ok, #state{}}.

handle_call({create, Session}, _From, State) ->
  %%判断是否是重连进程 如果是的话断掉之前的进程,重新插入
  case ets:lookup(?SESSION_TAB, Session#himq_session.client) of
    [S | _Tail] ->
      lager:info("find old session ~p", [S]),
      exit(S#himq_session.rec_pid, {shutdown, reconn});
    _ -> ok
  end,

  ets:insert(?SESSION_TAB, Session),
  lager:info("add session ~p sucess", [Session]),

  online(Session),
  {reply, ok, State};

handle_call(_Request, _From, State) ->
  {reply, ok, State}.

handle_cast({delete, Client, Reason}, State) ->
  lager:info("delete Client ~p", [Client]),
  case ets:lookup(?SESSION_TAB, Client) of [
    S | _Tail] ->
    ets:delete(?SESSION_TAB, Client),
    offline(S, Reason);
    _ -> ok end,
  {noreply, State};

handle_cast(_Request, State) ->
  {noreply, State}.

handle_info(_Info, State) ->
  {noreply, State}.

terminate(_Reason, _State) ->
  ok.
code_change(_OldVsn, State, _Extra) ->
  {ok, State}.

%%添加session
add(Mod,Protocol,Pv,Socket, Client, Type, Remark, ReadID, SendID) ->
  S = #himq_session{client = Client,socket = Socket, type = Type, remark = Remark, mod = Mod,protocol = Protocol,pv = Pv,
    rec_pid = ReadID, send_pid = SendID, add_time = himq:timestamp(), status = online},
  gen_server:call(?MODULE, {create, S}, 5000).

del(Client, Reason) -> gen_server:cast(?MODULE, {delete, Client, Reason}).


session_list(Page, PageSize) ->
  QH = qlc:q([Session || Session <- ets:table(?SESSION_TAB)]),
  Qc = qlc:cursor(QH),
  case Page of
    1 -> skip;
    _ -> qlc:next_answers(Qc, (Page - 1) * PageSize)
  end,
  case qlc:next_answers(Qc, PageSize) of
    [] ->
      #{<<"page">>=>Page,
        <<"count">>=>ets:info(?SESSION_TAB, size),
        <<"items">>=>[]
      };

    L ->
      R = lists:map(fun(E) ->
      Socket = E#himq_session.socket,
      lager:info("socket peername:~p,socket name ~p",[ranch_tcp:peername(Socket),ranch_tcp:sockname(Socket)]),

      {ok,{{A1,A2,A3,A4},Port1}}= ranch_tcp:peername(Socket),
      {ok,{{B1,B2,B3,B4},Port2}}= ranch_tcp:sockname(Socket),

      PeerName = list_to_binary(lists:concat([A1,".",A2,".",A3,".",A4,":"]++[Port1])),
      SocketName = list_to_binary(lists:concat([B1,".",B2,".",B3,".",B4,":"]++[Port2])),
      lager:info("socket stat:~p",[inet:getstat(Socket)]),
      {ok,StatList} = inet:getstat(Socket),

      lager:info("socket peername:~p,socket name ~p",[PeerName,SocketName]),
      Mod = E#himq_session.mod,
      Remark = Mod:session_remark(E#himq_session.remark),
      #{
        <<"client">>=>list_to_binary(E#himq_session.client),
        <<"created_at">>=>E#himq_session.add_time,
        <<"type">>=>E#himq_session.type,
        <<"pv">>=>E#himq_session.pv,
        <<"protocol">>=>E#himq_session.protocol,
        <<"peername">>=>PeerName,
        <<"listener">>=>SocketName,
        <<"stat">>=>jsx:encode(StatList),
        <<"remark">>=>Remark
      }
                       end, L),

      #{<<"page">>=>Page,
        <<"count">>=>ets:info(?SESSION_TAB, size),
        <<"items">>=>R
      }
  end.

ensure_table() ->
  case ets:info(?SESSION_TAB, size) of
    undefined ->
      ets:new(?SESSION_TAB, [set, public, named_table,
        {write_concurrency, true}, {read_concurrency, true},
        {keypos, #himq_session.client}]);
    S when is_integer(S) -> ok
  end.

format(S) ->
  #{
    <<"protocol">>=>atom_to_binary(S#himq_session.type, utf8),
    <<"time">>=>erlang:timestamp(),
    <<"id">>=>list_to_binary(S#himq_session.client)
  }.

%%链接断开
offline(S, Reason) ->
  link_log(S,?OFFLINE,Reason),
  Map = format(S),
  Packet = jsx:encode(Map#{<<"reason">>=>himq:to_binary(Reason)}),
  Pid = himq_mqtt_router:check_router(),
  %% 发送will消息
  send_will(S#himq_session.remark,Pid),
  himq_mqtt_router:dispatch(Pid, himq_session, {0, 1, 0, ?TOPIC_OFFLINE, Packet}).

send_will(#{wf:=1,will:=#{wr:=WRetain, wqos:=WQos, wtopic:=Topic, wmsg:=Packet}},Pid)->
  himq_mqtt_router:dispatch(Pid, himq_session, {0, WQos, WRetain, Topic, Packet});

send_will(#{wf:=0},_Pid)-> ok.

online(S) ->
  link_log(S,?ONLINE,login),
  Packet = jsx:encode(format(S)),
  Pid = himq_mqtt_router:check_router(),
  himq_mqtt_router:dispatch(Pid, himq_session, {0, 1, 0, ?TOPIC_ONLINE, Packet}).

link_log(Session,Status,Reason)->
  Client = himq:to_binary(Session#himq_session.client),
  Protocol=himq:to_binary(Session#himq_session.protocol),
  Item = #himq_link{id = Client,protocol = Protocol,created_at = himq:timestamp(),status = Status,remark = Reason},
  himq_mnesia:write(Item).

push(Push)->
  Client = Push#himq_push.client,
  Uuid = Push#himq_push.id,
  case ets:lookup(?SESSION_TAB, binary_to_list(Client)) of
    []->
      himq_mnesia:write(Push#himq_push{id=Uuid,status =offline}),
      himq_admin:msg(0,<<"设备已离线"/utf8>>);
    [S | _Tail] ->
      Pid = S#himq_session.send_pid,
      case is_process_alive(Pid) of
        true->
          Pid !{push,Push},
          himq_mnesia:write(Push#himq_push{id=Uuid,status =send}),
          himq_admin:msg(0,<<"消息已发送"/utf8>>);
        fase->
          himq_mnesia:write(#himq_push{id=Uuid,status =dead}),
          himq_admin:msg(0,<<"设备已离线"/utf8>>)
      end;
    E ->
      himq_mnesia:write(#himq_push{id=Uuid,status =exception}),
      lager:error("push msg when find session ~p",[E]),
      himq_admin:msg(1,<<"未知错误"/utf8>>)
  end.